chore: merge, ingest and store telemetry - #5146

todo: Could every received batch and terminal document outcome update docsDropped, batches, and batchesWithDrops before each return?

if you do maybe like an invalid batch test/priobe of docsDropped=1 while batches=0 and batchesWithDrops=0 the same early return and single document paths i think might omit the denominator and outcome counters?

You're right, i get docsDropped=1, batches=0, batchesWithDrops=0 from that. it also happens when every document was skipped as already merged. and batchesWithDrops only goes up when the merge errors, so the drop paths before storage never set it.

Fixed it, the batch is counted when it arrives, batchesWithDrops comes from the batch's own drop count, and every exit on the single-document path records a drop or a skip. enqueue also returns without running the handler when the document is in flight or the queue is full.

The doc-sync path recorded nothing either, so it does now, and db.Merge counts its conflicts and drops.

Test asserts merged + dropped + skipped equals what arrived.

todo: Could mergeDropReason use stable sentinels or typed matching without constructing stack traces per drop?

These probably do multiple allocations for one retry-exhaustion classification which might somewhat contradict the new stats comment that counting does not allocate until reporting?

Good catch. It built the errors it compared against, and constructing one captures a stack trace and renders it immediately. Fixed it so it matches package sentinels now, ErrCanNotIndexNonUniqueFields and a new client.ErrMaxTxnRetries.

It still allocates. defraError.Is compares Error() strings when the target is not a defraError, so the ipld.ErrNotFound case costs. The type comment now states what counting costs rather than claiming nothing allocates.

Cancellation and deadline were landing in other, they have their own contextDone reason now.

suggestion: Could skipDoc use a separate map or a document outcomes report label instead of reportFailureReasons("document drops", ...)? seems like skipDocstoresskip:alreadyMergedin the same map drained byreportFailureReasons("document drops", ...)`.

Implemented this, skips now go into their own map and get printed on their own document skips line.

The running totals were already separate, docsDropped and docsSkipped are different fields on the stats line. The problem was only the per-reason breakdown. Skip reasons went into the drop map with a skip: prefix, and when the reporter drained that map it printed all of them under one heading, document drops. So anyone reading that line, or grouping on it, saw skip:alreadyMerged sitting next to real losses.

There are tests for the split, and for a report draining both maps.

question: Could incumbentDocID and HeldBy be gated by read authorization or limited to internal merge diagnostics?

I went with keeping it on the node rather than gating it.

As far as I understand gating on read authorization wouldn't not work here, because the merge path has no caller identity to check against. I believe that is the same reason getDocForMerge already reads without the ACP filter (please correct me if I'm wrong). So now HeldBy is off the error entirely and newUniqueIndexError is back to the shape it had before this PR. The holder's docID now goes only to this node's log, next to the docID of the write that lost. incumbentDocID had a single caller, so its body is inlined into saveUniqueKey

suggestion: Could mergeChunk suppress duplicate merge chunk exhausted its retries logs and count the logical chunkExhausted outcome once across the initial chunk and one-event isolation retries?

Implemented both, the counter and the log now fire only when the chunk holds more than one event, which is the case that has a smaller write set to fall back to. That covers the isolation re-runs, since the isolation loop always passes a single-event slice.

It also means a single-event chunk's own first exhaustion is no longer counted or logged, including the trailing chunk when a batch is not a multiple of the chunk size. When that event then fails it is recorded as a retryExhausted drop, so it is not lost, it just is not counted as a chunk.

The drop accounting on this path did not change. An event that exhausts on its own is still counted under retryExhausted.

suggestion: Could fixed reason counters avoid the allocation from "skip:" + reason and the shared failureReasons mutex on skipDoc("alreadyMerged")?

i think right now there is a one allocation per duplicate call, and all workers likely share the failureReasons mutex.

I removed the allocation. Skips have their own map now, so the skip: prefix is no longer needed, and with no prefix there is no string to build. skipDoc measures zero allocations per call now

I did not fix the mutex. Drops and skips no longer share one, but that is two locks rather than the per-reason counters afaik. Every skip still takes the skip map's lock, and the callers are not one pool: the pubsub workers, the sync queue workers, and a goroutine per replicator stream. please correct me if I'm wrong

suggestion: Could direct DB.Merge and MergeBatchWithTxn share one exhaustion accounting path that preserves chunkConflicts and markExhausted() semantics?

I tried a manual direct-merge test it showed me a chunkConflicts=5 but chunkExhausted=0 after direct retries were exhausted, the incremental fix seemss to have corrected only the conflict counter, worth looking into imo.

You are right, it was inconsistent: mergeChunk counted any exhausted chunk including a single-event one, DB.Merge counted none.

Fixed by narrowing mergeChunk rather than adding to DB.Merge. A chunk of several is retried one event at a time and may still land everything. A single event has nothing smaller to retry, so it is a drop. One counter for both would hide that difference.

Both paths do record retryExhausted, on the merge drops line rather than the stats line.

Renamed chunkConflicts to txnConflicts, exhausted to chunkExhausted, and markExhausted() is gone.

question: Correct me if wrong, but atm seems like collectDAGBlocks inserts a CID into blockCIDs before a failed load increments missing? then buildCAR reads that same CID and returns carFailure(reasonBlockRead, err).

meaning the carMissingLinks counter therefore advances for a CAR that is never built, contrary to the short-CAR documentation.

suggestion: Could the skipped CID be excluded from the write set, or should this be reported as CAR failure?

You are right about the trace, but afaik excluding the CID would ship a short CAR. A receiver imports a CAR rather than walking the DAG, so a block missing from it never gets fetched, and the merge then fails on that link and drops the document. Reporting it as a CAR failure is what already happens, since the write loop cannot read the block either.

I moved the Add below the write loop, and a link is counted only for a CAR that gets returned. Both directions are covered: an abandoned CAR counts no missing link and no build, and a CAR that ships with two unfollowable links counts one build and two links.

suggestion: looks like getCollectionFromCollectionID can fail while creating a transaction or reading collections, but this new branch records every error as collectionNotFound. Could only the typed not-found error use collectionNotFound, with storage, transaction, cancellation, and unknown failures classified separately?

You are right. Fixed it so that only a real not-found error gets collectionNotFound, and everything else falls through to the normal drop classifier. That applies at both call sites, the direct merge and the batch one.

It does not give you all four categories tho, cancellation has its own contextDone reason now, and that covers the transaction case too, since opening a transaction can only fail with the database context's error.

question: Was going through this file and wondering does this mean that an in-flight duplicate returns nil immediately, and the branch maps the not-handled state to skipDoc("inFlight") before the active request's result is known. will the first request run later returning an error, so the same CID can be reported as one drop and one skip. Could duplicates inherit the active result, or should the counters be explicitly defined per unique CID?

Yes, correct. The duplicate returns nil straight away, and the same head does end up as one drop and one skip when the running arrival fails. I ran that case to be sure.

The counters record arrivals rather than unique CIDs. Two deliveries are two arrivals: one did the work and lost the document, the other did nothing. Since drops and skips are separate totals, the duplicate does not inflate the loss count.

Inheriting means blocking on the active result, and the second arrival still did no work. The branch also picks skip or drop from the returned error, so inheriting a non-nil result would record a syncQueueFull drop. Keying per CID would hide how often duplicates arrive.

question: batch branch bypasses the keyed processQueue and passes the untrusted req.Documents slice to MergeBatchWithTxn, whose contract requires independent document and collection keys. Repeated CIDs can therefore be reported merged and relayed more than once. Could the batch be deduplicated or rechecked under the same keyed merge gate?

You are right. As far as I understand, req.Documents does not reach MergeBatchWithTxn directly, processBatchedDocuments filters it into a fresh slice first. That slice is never deduplicated tho, so the repeats survive.

Rechecking under the keyed gate would not help if I understand correctly, enqueue blocks until its handler returns, so two copies in one request take the key in turn rather than at once, and the second still sees the block as unmerged, since the to-merge marker is only cleared by the merge itself. What saves the single-document path is that db.Merge runs inside the handler.

Stored data is unaffected, the rootstore is byte-identical after one merge and after two. The cost is the report and the relay. docsMerged counts the copy, and if the duplicate is of a create the second lands in updates, so one document shows in both. Each copy runs all of SendUpdate, so two pushes per replicator and two batch entries.

I have not fixed this yet. Caller side is the smaller change and lets me count the collapsed copy. Callee side covers every future caller, but then we have to pick what counts as a duplicate: on docIDs and collectionIDs it would drop a second genuine head, on head CID it removes only exact repeats. I would do the callee on head CID, returning true for the collapsed copy. Any recommendation or thoughts about this?

suggestion: Could importCAR pass the number of blocks written into carImportFailure and report it with the failure?atm telemetry exposes carImports=1 and carImportFailed=1 totals but no written-block count.

Added this. It now reports as carImportOrphanBlocks, with blocksWritten on the first log line for each reason.

It counts the blocks the import added, not the ones it tried to store. A CAR usually carries blocks we already hold, and Put returns nil either way, so the count comes from a Has first.

nit: I think returning missing instead of passing it as argument is more idiomatic / clean.

Add newFailureReasons() failureReasons that returns object with initialized maps. There is no reason to check on each record* call, with mutex locked. It's also not consistent with how P2P is initialized - with all the maps allocated in constructor. This change would break the nil-safe usage, of failureReasons, especially for tests, but we just need small helpers to initialize P2P instead of using &P2P{} in tests.

I completely don't know what to think about this method and how it's used in the code (same for carImportFailure). It works, it's smart, it's very compact on call, but feels very strange. Like a side effect to error.

I prefer how syncDAGFailure is implemented and used. It's more explicit, needs 1 more line of code for every call, but error handling feels more idiomatic.

Again, this should be in constructor, not on every call to markDropped.

Again, this could be returned among error.

Another constant.

And another constant

And the last constant from this file.

Oh, here is the type helper function to get the reason string from error, just like I suggested for loadBlockLinks/syncDAGFailure.

nit: created instead of isCreate (in all places).